#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>        //进程
#include "MQTTClient.h"     //MQTT
#include "cJSON.h"          //json
#include "msg_queue_peer.h" //消息队列
#include "shmem.h"          //共享内存
#include <sqlite3.h>        //数据库
#include <time.h>           //sleep

#define TOPIC "/app/data/down" //订阅
#define TOPIC2 "/app/data/up"  //发布
#define TIME 5                //自动保存时间

//数据库与错误信息定义
sqlite3 *db;
char *errmsg;
//共享内存
int *shmemNum = NULL;             //共享内存的数据量
struct std_node *shmemStd = NULL; //共享内存中首个元素地址
cJSON *dataList;
//MQTT
#define CLIENTID "ExampleClientPub"
#define QOS 1
#define TIMEOUT 10000L
volatile MQTTClient_deliveryToken deliveredtoken;
MQTTClient_deliveryToken token;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient client;
char ADDRESS[32];
int PERIOD = 0;

int key_his = 0;
pthread_t tid_his;

//MQTT回调函数
void connlost(void *context, char *cause);
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message);
void delivered(void *context, MQTTClient_deliveryToken dt);
//数据处理后使用mqtt_send函数发送到上位机
void sendAllData();
//MQTT发送
void mqtt_send(char *p);
//向设备发送key和value
void setDev(cJSON *key, cJSON *val);
//获取历史记录
void *getHistory(void *arg);
//写数据
void setAllList(int mod, int period);
//定时发送
void *up_period(void *arg);
//变换发送
void *up_vary(void *arg);

union val_t {
    int b_val;   //bool类型存储空间
    int i_val;   //整形值存储空间
    float f_val; //浮点值存储空间
};

struct std_node
{
    int key;             //唯一键值
    int type;            //数据点类型
    int dev_type;        //数据点属于哪个设备，根据网关支持的设备自行定义
    union val_t old_val; //变化上报后需要更新旧值
    union val_t new_val; //从共享内存取出最新数据，放到new_val中
    int ret;             //默认为-1，采集成功后设置为0，采集失败再置-1
};

struct msg
{
    long device_type; // 消息类型，用于区分32与modbus设备
    int key;
    union val_t val;
};

int main(int argc, char *argv[])
{
    time_t raw_time;
    //打开读取json文件
    FILE *fp;
    pthread_t tid;
    char buf[2048] = "";
    fp = fopen("./node.json", "r");
    if (fp == NULL)
    {
        perror("fopen err");
        return -1;
    }
    fread(buf, 2048, 1, fp);
    fclose(fp);
    //建立数据库与表
    if (sqlite3_open("./History.db", &db) != SQLITE_OK)
    {
        fprintf(stderr, "open err:%s", sqlite3_errmsg(db));
        return -1;
    }
    if (sqlite3_exec(db, "CREATE TABLE history(key INTEGER,val TEXT,time TEXT);", NULL, NULL, &errmsg))
        fprintf(stderr, "creat user err:%s\n", errmsg);
    //建立共享内存
    struct shm_param para;
    if (shm_init(&para, "SHMEM", 512) < 0)
    {
        perror("init err");
        return 0;
    }
    shmemNum = shm_getaddr(&para);
    printf("%p\n",shmemNum);
    if (shmemNum == NULL)
    {
        perror("getaddr err");
        return 0;
    }
    shmemStd = (struct std_node *)(shmemNum + 1);
    //解析点表
    dataList = cJSON_Parse(buf); //反序列化
    cJSON *Stm = cJSON_GetObjectItem(dataList, "stm32");
    cJSON *stmData = cJSON_GetObjectItem(Stm, "data");
    cJSON *Modbus = cJSON_GetObjectItem(dataList, "modbus");
    cJSON *modbusData = cJSON_GetObjectItem(Modbus, "data");
    *shmemNum = cJSON_GetArraySize(stmData) + cJSON_GetArraySize(modbusData); //确定表中数据个数
    struct std_node *temp = shmemStd;                                         //指向数据首地址
    for (int i = 0; i < cJSON_GetArraySize(stmData); i++)
    {
        cJSON *item = cJSON_GetArrayItem(stmData, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        temp->key = key->valueint;
        cJSON *type = cJSON_GetObjectItem(item, "type");
        temp->type = type->valueint;
        temp->dev_type = 0;
        temp->ret = -1;
        temp++;
    }
    for (int i = 0; i < cJSON_GetArraySize(modbusData); i++)
    {
        cJSON *item = cJSON_GetArrayItem(modbusData, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        temp->key = key->valueint;
        cJSON *type = cJSON_GetObjectItem(item, "type");
        temp->type = type->valueint;
        temp->dev_type = 1;
        temp->ret = -1;
        temp++;
    }
    temp = NULL;

    //MQTT连接
    cJSON *mqtt_server = cJSON_GetObjectItem(dataList, "mqtt_server");
    cJSON *addr = cJSON_GetObjectItem(mqtt_server, "addr");
    cJSON *port = cJSON_GetObjectItem(mqtt_server, "port");
    sprintf(ADDRESS, "tcp://%s:%d", addr->valuestring, port->valueint);
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    MQTTClient_create(&client, ADDRESS, CLIENTID,MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20; //心跳包检测
    conn_opts.cleansession = 1;       //MQTT连接检测
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, TOPIC, QOS); //订阅主题

    //默认上报模式
    cJSON *report = cJSON_GetObjectItem(dataList, "report");
    cJSON *up_mod = cJSON_GetObjectItem(report, "type");
    if(up_mod->valueint == 0) //
    {
        printf("当前上报模式：不上报\n");
    }else if(up_mod->valueint == 1) // vary 
    {
        printf("当前上报模式：变化上报\n");
        pthread_create(&tid, NULL, up_vary, NULL);
 
    }else if(up_mod->valueint == 2) // period
    {
        printf("当前上报模式：周期上报\n");
        cJSON *period = cJSON_GetObjectItem(report, "period");
        PERIOD=period->valueint;
        pthread_create(&tid, NULL, up_period, NULL);
    }
    while (1)
    {
        sleep(TIME);
        time(&raw_time);
        struct tm *time_info = localtime(&raw_time);
        char *time = asctime(time_info);
        char sql[128] = "";
        for (int i = 0; i < *shmemNum; i++)
        {
            char val[8] = "";
            if((shmemStd + i)->type == 1 || (shmemStd + i)->type == 2)
                sprintf(val, "%d", (shmemStd + i)->new_val.i_val);
            else if((shmemStd + i)->type == 3)
                sprintf(buf, "%.1f", (shmemStd + i)->new_val.f_val);
            sprintf(sql, "INSERT INTO History VALUES (%d,'%s','%s');", (shmemStd + i)->key, val, time);
            sqlite3_exec(db, sql, NULL, NULL, &errmsg);
            printf("插入完毕\n");
        }
    }
    return 0;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("收到数据\n");
    int i;
    char *payloadptr;
    payloadptr = message->payload;
    cJSON *root = cJSON_Parse(payloadptr);
    if (NULL == root)
    {
        printf("parse err\n");
        return -1;
    }
    cJSON *type = cJSON_GetObjectItem(root, "type");
    if (type->valueint == 1)//刷新上报
    {
        sendAllData();
    }
    else if (type->valueint == 2)//操控设备
    {
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        cJSON *val = cJSON_GetObjectItem(data, "val");
        setDev(key, val);
    }
    else if (type->valueint == 3)//定时上报
    {
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *type_mod = cJSON_GetObjectItem(data, "type");
        cJSON *type_period = cJSON_GetObjectItem(data, "period");
        setAllList(type_mod->valueint, type_period->valueint);
    }
    else if (type->valueint == 4)//历史记录查询
    {
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        key_his = key->valueint;
        pthread_create(&tid_his, NULL, getHistory, NULL);
    }
    cJSON_Delete(root);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    deliveredtoken = dt;
}

void sendAllData()
{
    cJSON *root = cJSON_CreateObject();
    cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
    cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
    cJSON *arr = cJSON_CreateArray();
    cJSON_AddItemToObject(root, "data", arr);
    for (int i = 0; i < *shmemNum; i++)
    {
        cJSON *data1 = cJSON_CreateObject();
        cJSON_AddItemToArray(arr, data1);
        char buf[8] = "";
        if ((shmemStd + i)->type == 1 || (shmemStd + i)->type == 2)
            sprintf(buf, "%d", (shmemStd + i)->new_val.i_val);
        else
            sprintf(buf, "%.1f", (shmemStd + i)->new_val.f_val);
        cJSON_AddItemToObject(data1, "key", cJSON_CreateNumber((shmemStd + i)->key));
        cJSON_AddItemToObject(data1, "val", cJSON_CreateString(buf));
    }
    char *p = cJSON_Print(root);
    mqtt_send(p);
    printf("上报\n");
    free(p);
    cJSON_Delete(root);
}

void mqtt_send(char *p)
{
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    pubmsg.payload = p;
    pubmsg.payloadlen = (int)strlen(p);
    MQTTClient_publishMessage(client, TOPIC2, &pubmsg, &token);
    printf("数据已发送\n");
}

void setDev(cJSON *key, cJSON *val)
{
    struct msg msg_set;
    msg_set.key = key->valueint;
    for (int i = 0; i < *shmemNum; i++)
    {
        if ((shmemStd + i)->key == key->valueint)
        {
            if ((shmemStd + i)->dev_type == 0)
            {
                msg_set.device_type = 3;
            }
            else if ((shmemStd + i)->dev_type == 1)
            {
                msg_set.device_type = 1;
            }
            char *ch = val->valuestring;
            msg_set.val.b_val = *ch - '0';
            printf("key=%d\n", msg_set.key);
            printf("type=%ld\n", msg_set.device_type);
            printf("val=%d\n", msg_set.val.b_val);
            msg_queue_send("MSG", &msg_set, sizeof(msg_set), 0); //将数据放在消息队列中，其中包括消息队列的初始化
        }
    }

    printf("setDev ok\n");
}

void *getHistory(void *arg)
{
    char sql[68] = "";
    sprintf(sql, "SELECT * FROM History WHERE key=%d;", key_his);
    int KEY = key_his;
    char **resultp;
    int n_row, n_cloum;
    int i, j;
    if (sqlite3_get_table(db, sql, &resultp, &n_row, &n_cloum, &errmsg) != SQLITE_OK)
    {
        printf("err:%s\n", errmsg);
        return NULL;
    }
    char **p;
    p = resultp + n_cloum;
    char data[64] = "";
    for (i = 0; i < n_row; i++)
    {
        sprintf(data, "key:%s  val:%s time:%s", p[(i * n_cloum) + 0], p[(i * n_cloum) + 1], p[(i * n_cloum) + 2]);
        cJSON *root = cJSON_CreateObject();
        cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(4));
        cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
        cJSON_AddItemToObject(root, "key", cJSON_CreateNumber(KEY));
        cJSON_AddItemToObject(root, "data", cJSON_CreateString(data));
        char *p = cJSON_Print(root);
        mqtt_send(p);
        free(p);
    }
    printf("历史记录: %d条\n", n_row);
}

void setAllList(int mod, int period)
{
    printf("setAllList ok\n");
    cJSON *Report = cJSON_GetObjectItem(dataList, "report");
    cJSON_ReplaceItemInObject(Report, "type", cJSON_CreateNumber(mod));
    cJSON_ReplaceItemInObject(Report, "period", cJSON_CreateNumber(period));
    char *p = cJSON_Print(dataList);
    FILE *fp;
    fp = fopen("./node.json", "w");
    if (fp == NULL)
    {
        perror("fopen err");
        return;
    }
    fwrite(p, strlen(p), 1, fp);
    fclose(fp);
}

void *up_period(void *arg) 
{
    while (1)
    {
        sleep(PERIOD);
        sendAllData();
    }
}
 
void *up_vary(void *arg)
{
    cJSON *root = cJSON_CreateObject();
    cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
    cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
    cJSON *arr = cJSON_CreateArray();
    cJSON_AddItemToObject(root, "data", arr);
    cJSON *data1 = cJSON_CreateObject();
    cJSON_AddItemToArray(arr, data1);
    cJSON_AddItemToObject(data1, "key", cJSON_CreateNumber(0));
    cJSON_AddItemToObject(data1, "val", cJSON_CreateString(""));
 
    while (1)
    {
        for(int i =0 ;i<*shmemNum;i++)
        {
            if(memcmp(&(shmemStd+i)->old_val,&(shmemStd+i)->new_val ,sizeof(union val_t)) != 0)
            {
                char buf[8]="";
                if( (shmemStd+i)->type==1 || (shmemStd+i)->type==2 )//int,bool
                {
                    sprintf(buf,"%d",(shmemStd+i)->new_val.i_val);
                    printf("new:%d\n",(shmemStd+i)->new_val.i_val);
                    printf("old:%d\n",(shmemStd+i)->old_val.i_val);
                }
                else 
                {
                    sprintf(buf,"%.1f",(shmemStd+i)->new_val.f_val);
                    printf("new:%f\n",(shmemStd+i)->new_val.f_val);
                    printf("old:%f\n",(shmemStd+i)->old_val.f_val);
                }
                cJSON_ReplaceItemInObject(data1,"key",cJSON_CreateNumber((shmemStd+i)->key));
                cJSON_ReplaceItemInObject(data1,"val",cJSON_CreateString(buf));
                char *p = cJSON_Print(root);
                mqtt_send(p);
                free(p);
                (shmemStd+i)->old_val.i_val = (shmemStd+i)->new_val.i_val;
                printf("数据有变化\n");
            }
        }
    }
    cJSON_Delete(root);
}